package com.study.flink.source

import java.util.Properties

import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.streaming.api.scala.{DataStream, StreamExecutionEnvironment}
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer

/**
  * kafka source demo
  *
  * @author stephen
  * @date 2019-07-22 11:46
  */
object ScalaKafkaSourceDemo {

  def main(args: Array[String]): Unit = {
    // 1 获取执行环境
    val env = StreamExecutionEnvironment.getExecutionEnvironment

    // 2 获取kafka数据
    val properties: Properties = new Properties()
    properties.setProperty("bootstrap.servers", "localhost:9092")
    properties.setProperty("group.id", "consumer-group")
    properties.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")
    properties.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer")
    properties.setProperty("auto.offset.reset", "latest")

    val consumer:FlinkKafkaConsumer[String] = new FlinkKafkaConsumer[String]("test-topic",new SimpleStringSchema(), properties)
    import org.apache.flink.api.scala._
    val dataStream:DataStream[String] = env.addSource(consumer)

    // 3 transformation
    val wordStream = dataStream.flatMap(_.split(" "))

    // 4 输出
    wordStream.print()

    // 5 启动任务
    env.execute("Kafka source demo")

  }
}
